package com.whut.domain;

import com.whut.util.FileUtil;
import com.whut.util.JSchUtil;
import com.whut.util.OSUtil;
import com.whut.util.SendMessageUtil;
import com.whut.wesocket.PollWebSocketServlet;
import org.apache.log4j.Logger;
import org.json.JSONObject;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.text.ParseException;

/**
 * Created by YY on 2018-01-21.
 */
public class SparkUser implements Runnable{

    private static final Logger logger = Logger.getLogger(SparkUser.class);
    private String basePath;
    private String username;
    private String userPath;
    private PollWebSocketServlet webSocketServlet;
    private String email;
    private String md5;

    public static Logger getLogger() {
        return logger;
    }

    public String getMd5() {
        return md5;
    }

    public void setMd5(String md5) {
        this.md5 = md5;
    }

    //设置跑某个python的路径以及使用的资源
    private String cpu;
    private String memory;
    private String projectName;
    private String runFilePath;

    public String getUsername() {
        return username;
    }
    public void setUsername(String username) {
        this.username = username;
    }
    public String getUserPath() {
        return userPath;
    }
    public void setUserPath(String userPath) {
        this.userPath = userPath;
    }
    public PollWebSocketServlet getWebSocketServlet() {
        return webSocketServlet;
    }
    public String getEmail() {
        return email;
    }
    public String getCpu() {
        return cpu;
    }
    public void setCpu(String cpu) {
        this.cpu = cpu;
    }
    public String getMemory() {
        return memory;
    }
    public void setMemory(String memory) {
        this.memory = memory;
    }
    public String getProjectName() {
        return projectName;
    }
    public void setProjectName(String projectName) {
        this.projectName = projectName;
    }
    public String getRunFilePath() {
        return runFilePath;
    }
    public void setRunFilePath(String runFilePath) {
        this.runFilePath = runFilePath;
    }
    public void setWebSocketServlet(PollWebSocketServlet webSocketServlet) {
        this.webSocketServlet = webSocketServlet;
    }
    public void setEmail(String email) {
        this.email = email;
    }
    public SparkUser(String username)
    {
        this.username=username;
        if(OSUtil.isWindow()) {
            basePath="C:\\Users\\Administrator\\Desktop";
            userPath=basePath+"\\"+username;
        }
        else {
            basePath="/home/ubuntu/spark";
            userPath=basePath+"/"+username;
        }
    }
    public SparkUser(String username,String email) {
        this.username=username;
        this.email=email;
        if(OSUtil.isWindow())
            userPath="C:\\Users\\Administrator\\Desktop\\"+username;
        else
            userPath="/home/ubuntu/spark/"+username;
    }
    public int createProject(String projectName)
    {
        return FileUtil.newFolder(userPath+"/"+projectName);
    }
    public boolean deleteProject(String projectName)
    {
        return FileUtil.delFolder(userPath+"/"+projectName);
    }
    public int createFolder(String relativePath) {
        if(relativePath.startsWith("/")) {
            return FileUtil.newFolder(basePath+relativePath);
        }
        else {
            return FileUtil.newFolder(basePath+"/"+relativePath);
        }
    }
    public boolean deleteFolder(String relativePath){
        if(relativePath.startsWith("/")) {
            return FileUtil.delFolder(basePath+relativePath);
        }
        else {
            return FileUtil.delFolder(basePath+"/"+relativePath);
        }
    }
    public int createFile(String relativePath,String text) {
        if(relativePath.startsWith("/")) {
            return FileUtil.newFile(basePath+relativePath,text);
        }
        else {
            return FileUtil.newFile(basePath+"/"+relativePath,text);
        }
    }
    public int createFileByCover(String relativePath,String text) {
        if(relativePath.startsWith("/")) {
            return FileUtil.newFileByCover(basePath+relativePath,text);
        }
        else {
            return FileUtil.newFileByCover(basePath+"/"+relativePath,text);
        }
    }
    public int appendFile(String relativePath,String text) {
        if(relativePath.startsWith("/")) {
            return FileUtil.appendFile(basePath+relativePath,text);
        }
        else {
            return FileUtil.appendFile(basePath+"/"+relativePath,text);
        }
    }
    public boolean deleteFile(String relativePath)
    {
        if(relativePath.startsWith("/")) {
            return FileUtil.delFile(basePath+relativePath);
        }
        else {
            return FileUtil.delFile(basePath+"/"+relativePath);
        }
    }
    public String getFileText(String relativePath)
    {
        if(relativePath.startsWith("/")) {
            return FileUtil.readFileByLines(basePath+relativePath);
        }
        else {
            return FileUtil.readFileByLines(basePath+"/"+relativePath);
        }
    }
    public boolean clearFolder(String relativePath) {
        if(relativePath.startsWith("/")) {
            return FileUtil.delAllFile(basePath+relativePath);
        }
        else {
            return FileUtil.delAllFile(basePath+"/"+relativePath);
        }
    }
    public JSONObject getProjectSchema(String projectName) {
        return FileUtil.getDirectoryCatalog(userPath+"/"+projectName);
    }
    public JSONObject getAllProjectSchema() {
        return FileUtil.getDirectoryCatalog(userPath);
    }

    public int renameFile(String relativePath,String newName)
    {
        if(relativePath.startsWith("/")) {
            return FileUtil.renameFile(basePath+relativePath,newName);
        }
        else {
            return FileUtil.renameFile(basePath+"/"+relativePath,newName);
        }
    }
    private boolean copyProjectToMaster(String projectName)
    {
        try {
            SendMessageUtil.sendMessage(md5, "正在将文件分散到Spark集群...", webSocketServlet);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        JSchUtil jSchUtil=new JSchUtil("192.168.1.2","root","root");
        if(jSchUtil.execute("mkdir -p "+userPath)!=0) {
            return false;
        }
        boolean result=true;
        String command = "scp -r " + userPath+"/"+projectName + " root@192.168.1.2:" + userPath;
        try {
            Process pr = Runtime.getRuntime().exec(command);
            InputStream inputStream = pr.getErrorStream();
            BufferedReader in = new BufferedReader(new InputStreamReader(inputStream));
            String line;
            while ((line = in.readLine()) != null) {
                result=false;
                logger.info(line);
                SendMessageUtil.sendMessage(md5, line, webSocketServlet);
            }
            inputStream.close();
            in.close();
            pr.waitFor();
            pr.destroy();
        } catch (Exception e) {
            result=false;
            e.printStackTrace();
        }
        logger.info("成功复制项目给Master");
        return result;
    }
    private boolean deleteProjectInMaster(String projectName) {
        JSchUtil jSchUtil=new JSchUtil("192.168.1.2","root","root");
        return (jSchUtil.execute("rm -rf "+userPath+"/"+projectName)==0);
    }
    @Override
    public void run() {
        logger.info("deleteProjectInMaster:"+deleteProjectInMaster(projectName));
        boolean result = copyProjectToMaster(projectName);
        logger.info("copyProjectToMaster:"+result);
        if(!result)
            return;
        String command="cd /usr/local/spark\n" +
                "./bin/spark-submit " +
                basePath+"/"+runFilePath+" " +
                "--master yarn-cluster " +
                "--executor-memory "+memory+" G "+
                "--num-executors "+ cpu;
        JSchUtil jSchUtil=new JSchUtil("192.168.1.2","root","root");
        try {
            SendMessageUtil.sendMessage(md5, "开始提交作业到Spark集群...", webSocketServlet);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        jSchUtil.execute(command,md5,webSocketServlet);
    }
}
